package com.xrui.adc.hbase.codec;

import com.google.common.base.Preconditions;
import com.moz.kiji.delegation.NamedProvider;


import com.xrui.adc.hbase.hashing.RowKeyHasher;
import com.xrui.hbase.RowKeyComponents;
import com.xrui.hbase.codec.RowKeyDecoder;
import com.xrui.hbase.codec.RowKeyEncoder;
import com.xrui.hbase.hashing.Hasher;
import com.xrui.hbase.schema.RowKeyComponent;
import com.xrui.hbase.schema.RowKeyFormat;
import org.apache.commons.lang.ArrayUtils;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.util.StringUtils;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.regex.Pattern;

public class RowKeyCodec implements RowKeyDecoder, RowKeyEncoder, NamedProvider {
    public static final String PROVIDER_NAME = RowKeyCodec.class.getName();

    @Override
    public String getName() {
        return PROVIDER_NAME;
    }

    @Override
    public RowKeyComponents decode(byte[] bytes, RowKeyFormat format) throws IOException {
        int hashSize = format.getHashSize();
        String value = Bytes.toString(bytes);
        String componentString = value;
        if (format.isHashed()) {
            componentString = value.substring(hashSize);
        }

        List<RowKeyComponent> components = format.getComponents();
        Object[] items = new Object[] { componentString};
        if (components.size() > 1) {
            String delimiter = format.getDelimiter();
            if (delimiter == null || delimiter.isEmpty()) {
                throw new IOException(String.format("Delimiter is null or empty: %s", componentString));
            }
            items = componentString.split(Pattern.quote(delimiter));
            if (items.length != components.size()) {
                throw new IOException(
                    String.format("Expect %s fields but got %s: %s", components.size(), items.length, componentString)
                );
            }
        }

        return RowKeyComponents.fromComponents(items);
    }

    @Override
    public byte[] encode(RowKeyComponents components, RowKeyFormat format) throws IOException {
        Object[] componentArray = components.getComponents();
        Hasher hasher = RowKeyHasher.Factory.create(format.getHasher());

        List<String> items = Arrays.asList(Arrays.copyOf(componentArray, componentArray.length, String[].class));
        List<String> hashItems = new ArrayList<>();
        long hashItemSize = format.getComponents().stream().filter(RowKeyComponent::isHashed).count();
        Preconditions.checkArgument(
            componentArray.length >= hashItemSize,
            String.format("Expect %s hashed row key components, but got %s", hashItemSize, componentArray.length)
        );
        for (int index = 0; index < hashItemSize; index++) {
            hashItems.add(String.valueOf(componentArray[index]));
        }

        byte[] key = Bytes.toBytes(StringUtils.join(format.getDelimiter(), items));
        if (hashItems.isEmpty()) {
            return key;
        } else {
            byte[] hashBytes = hasher.getHashBytes(StringUtils.join(format.getDelimiter(), hashItems));
            return ArrayUtils.addAll(hashBytes, key);
        }
    }
}
